package com.atguigu.chapter05;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new Clicksource()).map( a -> Tuple2.of(a.user,1L)).returns(new TypeHint<Tuple2<String, Long>>() {}).
                //使用用户名来分流 注意泛型擦除
        keyBy(a -> a.f0).reduce( (a, b) -> Tuple2.of(a.f0,a.f1+b.f1) ).returns(new TypeHint<Tuple2<String, Long>>() {}).
                //为每一条数据分配同一个 key，将聚合结果发送到一条流中去
        keyBy(a -> true).reduce( (a,b) -> a.f1>b.f1? a : b ).returns(new TypeHint<Tuple2<String, Long>>() {}).print();

        env.execute();


    }
}
